package org.example.graphx.local

import org.apache.spark.graphx.GraphLoader
import org.apache.spark.sql.SparkSession

object PageRankExample {

  def main(args: Array[String]): Unit = {

    // 创建一个SparkSession.
    val spark = SparkSession
      .builder
      .master("local")
      .appName("graphx-pagerank")
      .getOrCreate()
    val sc = spark.sparkContext

    val filePath = "data/graphx/followers.txt";
    // 读取图数据文件
    val graph = GraphLoader.edgeListFile(sc, filePath)

    // 调用PageRank算法
    val ranks = graph.pageRank(0.0001).vertices

    // 将节点映射为人物名称
    val users = sc.textFile("data/graphx/users.txt").map { line =>
      val fields = line.split(",")
      (fields(0).toLong, fields(1))
    }
    val ranksByUsername = users.join(ranks).map {
      case (id, (username, rank)) => (username, rank)
    }

    // 在控制台输出结果
    println(ranksByUsername.collect().mkString("\n"))

    spark.stop()
  }

}
